package com.zishuimuyu.mqtt;

import com.zishuimuyu.config.MqttConfig;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author 向佩环
 * @description mqtt数据接收、回调
 * @company 北京木联能软件股份有限公司
 * @date 2022/9/7
 */
@Slf4j
@Component
public class MyMqttCallback implements MqttCallback {

    @Resource
    private MqttConfig mqttConfig;

    private MqttClient client;
    private MqttConnectOptions options;

    public MyMqttCallback() {
    }

    public void setClientAndOptions(MqttClient client, MqttConnectOptions options) {
        this.client = client;
        this.options = options;
    }


    /**
     * 连接丢失时被调用
     */
    @Override
    @SneakyThrows
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connectionLost 连接断开，准备重连: {}", throwable.getMessage());
        boolean reconnected = false;
        int attempt = 0;
        int maxAttempts = 10;

        while (!reconnected && attempt < maxAttempts) {
            attempt++;
            for (String url : mqttConfig.getUrls()) {
                try {
                    log.info("尝试重新连接到 URL: {}", url);
                    MqttClient newClient = new MqttClient(url, mqttConfig.getClientId(), new MemoryPersistence());
                    newClient.setCallback(this);
                    newClient.connect(options);
                    log.info("MQTT 成功连接 URL: {}", url);
                    client = newClient;
                    MyMqttClient.setClient(client);
                    reconnected = true;
                    break;
                } catch (MqttException e) {
                    log.warn("无法连接 URL: {}, 错误: {}", url, e.getMessage());
                }
            }
            if (!reconnected) {
                log.error("无法连接任何配置的 URL, 尝试次数: {}", attempt);
                if (attempt < maxAttempts) {
                    log.info("5 秒后重试...");
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.error("线程被中断: {}", ie.getMessage());
                    }
                } else {
                    log.error("已达到最大重连次数: {}", maxAttempts);
                }
            }
        }
    }

    /**
     * 接收到消息时被调用
     *
     * @param topic
     * @param mqttMessage
     *
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("MyMqttCallback:" + topic + ";msg:" + new String(mqttMessage.getPayload()));

    }

    /**
     * 消息发送完成时被调用
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
       /* try {
            log.info(String.valueOf(iMqttDeliveryToken.getMessage()));
        } catch (MqttException e) {
            e.printStackTrace();
        }*/
    }
}
